#include "PahoMqttAsyncClient.h"
#include <stdio.h>
#include <string.h>
#include <debug_printf.h>
#include "thmar_types.h"
#include <stdlib.h>
#include "local_error.h"

namespace smq {

typedef struct error_info {
    int code;
    const char* message;
} error_info_t;
// paho mqtt 错误码及其对应的错误信息
static error_info_t error_infos[] = {
    {MQTTASYNC_SUCCESS, "success"},
    {MQTTASYNC_FAILURE, "failure"},
    {MQTTASYNC_PERSISTENCE_ERROR, "persistence error"},
    {MQTTASYNC_DISCONNECTED, "disconnected"},
    {MQTTASYNC_MAX_MESSAGES_INFLIGHT, "max messages inflight"},
    {MQTTASYNC_BAD_UTF8_STRING, "bad utf8 string"},
    {MQTTASYNC_NULL_PARAMETER, "null parameter"},
    {MQTTASYNC_TOPICNAME_TRUNCATED, "topicname truncated"},
    {MQTTASYNC_BAD_STRUCTURE, "bad structure"},
    {MQTTASYNC_BAD_QOS, "bad qos"},
    {MQTTASYNC_NO_MORE_MSGIDS, "no more msgids"},
    {MQTTASYNC_OPERATION_INCOMPLETE, "operation incomplete"},
    {MQTTASYNC_MAX_BUFFERED_MESSAGES, "max buffered messages"},
    {MQTTASYNC_SSL_NOT_SUPPORTED, "ssl not supported"},
    {MQTTASYNC_BAD_PROTOCOL, "bad protocol"},
    {MQTTASYNC_BAD_MQTT_OPTION, "bad mqtt option"},
    {MQTTASYNC_WRONG_MQTT_VERSION, "wrong mqtt version"},
    {MQTTASYNC_0_LEN_WILL_TOPIC, "0 len will topic"},
    {MQTTASYNC_COMMAND_IGNORED, "command ignored"},
    {MQTTASYNC_MAX_BUFFERED, "max buffered"}
};

// 返回 paho mqtt 错误码对应的错误信息
static const char* get_error_message(int code) {
    for(int i=0; i<sizeof(error_infos)/sizeof(error_info_t); i++) {
        if(error_infos[i].code == code) {
            return error_infos[i].message;
        }
    }
    return "unknown error";
}
// 根据 paho mqtt 错误码设置本地错误码及其对应的错误信息
static void set_local_error_message(int paho_code) {
    thmar_set_err(paho_code,get_error_message(paho_code));
}

static void connectionLost(void *context, char *cause)
{
    PahoMqttAsyncClient* client = (PahoMqttAsyncClient*)context;

    printf("Connection lost\n");
    if(cause)
        printf("cause: %s\n", cause);
    auto connectionOptions = client->getConnectOptions();
    connectionOptions.cleansession = 1;
    int code;
    if((code = client->connect()) == MQTTASYNC_SUCCESS)
    {
        printf("Failed to start connect, return code %d\n", code);
        exit(EXIT_FAILURE);
    }
    printf("Auto reconnect success\n");
}
static int messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message){
    PahoMqttAsyncClient* pahoClient = (PahoMqttAsyncClient*)context;
    if(pahoClient->channelDispatcher) {
        COM_PRINTF_DEBUG("call dispatcher function %p\n",pahoClient->channelDispatcher);
        pahoClient->channelDispatcher(pahoClient->dispatcherContext,topicName, (size_t)topicLen, (char*)message->payload, (size_t)message->payloadlen);
    }
    return 1;
}
static void deliveryComplete(void* context, MQTTAsync_token token){
    printf("Delivery complete\n");
}

static void onSubscribe(void *context, MQTTAsync_successData *response)
{
    printf("Subscribe successed\n");
}

static void onSubsribeFailure(void *context, MQTTAsync_failureData *response)
{
    printf("Subscribe failed, rc %d\n", response ? response->code : 0);
}
void onConnectFailure(void *context, MQTTAsync_failureData *response)
{
    printf("Connect failed, rc %d\n", response ? response->code : 0);
}
void onConnect(void *context, MQTTAsync_successData *response)
{
    printf("Successful connection\n");
}
static int init(PahoMqttAsyncClient* pahoClient)
{
    int code = MQTTAsync_create(&pahoClient->client, pahoClient->serverURI, "clientId", MQTTCLIENT_PERSISTENCE_NONE, nullptr);
    if(code != MQTTASYNC_SUCCESS){
        set_local_error_message(code);
        return THMR_MQTT_INIT_FAILED;
    }
    pahoClient->conn_opts.keepAliveInterval = 20;
    pahoClient->conn_opts.cleansession = 1;
    pahoClient->conn_opts.onSuccess = onConnect;
    pahoClient->conn_opts.onFailure = onConnectFailure;
    pahoClient->conn_opts.context = pahoClient->client;

    code = MQTTAsync_setCallbacks(pahoClient, pahoClient, connectionLost, messageArrived, deliveryComplete);
    if(code != MQTTASYNC_SUCCESS){
        set_local_error_message(code);
        return THMR_MQTT_INIT_FAILED;
    }
    return THMR_OK;
}
static void deinit(PahoMqttAsyncClient* pahoClient)
{
    MQTTAsync_disconnect(pahoClient, &pahoClient->disc_opts);
    MQTTAsync_destroy(&pahoClient->client);
    free((void*)pahoClient->serverURI);
    pahoClient->serverURI = nullptr;
    free((void*)pahoClient->clientID);
    pahoClient->clientID = nullptr;
    if(pahoClient->conn_opts.username){
        free((void*)pahoClient->conn_opts.username);
        pahoClient->conn_opts.username = nullptr;
    }
    if(pahoClient->conn_opts.password){
        free((void*)pahoClient->conn_opts.password);
        pahoClient->conn_opts.password = nullptr;
    }
}
PahoMqttAsyncClient::PahoMqttAsyncClient(MQTTAsync_connectOptions conn_opts,const char* serverURI,const char *clientID,int qos)
        :conn_opts(conn_opts),qos(qos),initialized(-1),disc_opts(MQTTAsync_disconnectOptions_initializer),channelDispatcher(nullptr) {
    this->serverURI = serverURI ? strdup(serverURI): strdup("tcp://localhost:1883") ;
    this->clientID = clientID ? strdup(clientID): strdup("ThmarMqttClient") ;
    if(this->conn_opts.username)
        this->conn_opts.username = strdup(this->conn_opts.username);
    if(this->conn_opts.password)
        this->conn_opts.password = strdup(this->conn_opts.password);
    this->initialized = init(this);
}

PahoMqttAsyncClient::PahoMqttAsyncClient(const char *serverURI, const char *username, const char *password, const char *clientID, int timeout, int keepAliveInterval, int qos)
{
    MQTTAsync_connectOptions conn_opts = { {'M', 'Q', 'T', 'C'}, 8, keepAliveInterval, 1, 65535, NULL, username, password, timeout, 0,\
        NULL, NULL, NULL, NULL, 0, NULL, MQTTVERSION_DEFAULT, 0, 1, 60, {0, NULL}, 0, NULL, NULL, NULL, NULL, NULL, NULL, NULL};
    PahoMqttAsyncClient(conn_opts, serverURI, clientID, qos);
}

PahoMqttAsyncClient::~PahoMqttAsyncClient() {
    deinit(this);
}

int PahoMqttAsyncClient::connect()
{
    int code= MQTTAsync_connect(client, &conn_opts);
    if (code != MQTTASYNC_SUCCESS)
    {
        set_local_error_message(code);
        return THMR_CONNECT_FAILED;
    }
    return THMR_OK;
}

void PahoMqttAsyncClient::disconnect() {
    MQTTAsync_disconnect(client, &disc_opts);
}

int PahoMqttAsyncClient::publish(const char* topic, const char* message) {
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    opts.context = this;
    MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
    pubmsg.payload = (void*)message;
    pubmsg.payloadlen = strlen(message);
    pubmsg.qos = qos;
    pubmsg.retained = 0;
    int code = MQTTAsync_sendMessage(client, topic, &pubmsg, &opts);
    if(code != MQTTASYNC_SUCCESS){
        set_local_error_message(code);
        return THMR_PUBLISH_FAILED;
    }
    return THMR_OK;
}

MQTTAsync_connectOptions& PahoMqttAsyncClient::getConnectOptions()
{
    return conn_opts;
}

int PahoMqttAsyncClient::subscribe(const char* topic) {
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    opts.context = this;
    opts.onSuccess = onSubscribe;
    opts.onFailure = onSubsribeFailure;
    int code = MQTTAsync_subscribe(client, topic, qos, &opts);
    if (code != MQTTASYNC_SUCCESS) {
        set_local_error_message(code);
        return THMR_SUBSCRIBE_FAILED;
    }
    return THMR_OK;
}

int PahoMqttAsyncClient::unsubscribe(const char *channel)
{
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    int code = MQTTAsync_unsubscribe(client, channel, &opts);
    if (code != MQTTASYNC_SUCCESS) {
        set_local_error_message(code);
        return THMR_UNSUBSCRIBE_FAILED;
    }
    return THMR_OK;
}

} // namespace smq